//package com.heima.article.stream;
//
//import com.alibaba.fastjson.JSON;
//import com.heima.common.constants.HotArticleConstants;
//import com.heima.model.mess.ArticleVisitStreamMess;
//import com.heima.model.mess.UpdateArticleMess;
//import lombok.extern.slf4j.Slf4j;
//import org.apache.commons.lang3.StringUtils;
//import org.apache.kafka.streams.KeyValue;
//import org.apache.kafka.streams.StreamsBuilder;
//import org.apache.kafka.streams.kstream.*;
//import org.springframework.context.annotation.Bean;
//import org.springframework.context.annotation.Configuration;
//
//import java.time.Duration;
//
//@Configuration
//@Slf4j
//public class HotArticleStreamHandler105 {
//
//    @Bean
//    public KStream<String,String> kStream(StreamsBuilder streamsBuilder){
//        //接收消息
//        KStream<String,String> stream = streamsBuilder.stream(HotArticleConstants.HOT_ARTICLE_SCORE_TOPIC);
//        //聚合流式处理
//        //1.2.来的这条消息key value 3.之前聚合之后的结果
//        stream.map((key,value)->{
//            UpdateArticleMess mess = JSON.parseObject(value, UpdateArticleMess.class);
//            //重置消息的key:1234343434   和  value: likes:1
//            return new KeyValue<>(mess.getArticleId().toString(),mess.getType().name()+":"+mess.getAdd());
//        })
//                //按照文章id进行聚合
//                .groupByKey()
//                //时间窗口
//                .windowedBy(TimeWindows.of(Duration.ofSeconds(10)))
//                /**
//                 * 自行的完成聚合的计算
//                 */
//
//                .aggregate(new Initializer<String>() {
//                    /**
//                     * 初始方法，返回值是消息的value
//                     * @return
//                     */
//                    @Override
//                    public String apply() {
//                        ArticleVisitStreamMess articleVisitStreamMess = new ArticleVisitStreamMess();
//                        return JSON.toJSONString(articleVisitStreamMess);
//                    }
//                    /**
//                     * 真正的聚合操作，返回值是消息的value
//                     */
//                }, (key, value, aggValue) -> {
//                    ArticleVisitStreamMess mess = JSON.parseObject(aggValue, ArticleVisitStreamMess.class);
//                    if(mess.getArticleId() == null){
//                        //第一次给文章ID赋值
//                        mess.setArticleId(Long.valueOf(key));
//                    }
//
//                    //value LIKES:1
//                    String[] valAry = value.split(":");  //["LIKES",1]
//                    switch (UpdateArticleMess.UpdateArticleType.valueOf(valAry[0])){
//                        case COLLECTION:
//                            mess.setCollect(mess.getCollect() + Integer.valueOf(valAry[1]));
//                            break;
//                        case COMMENT:
//                            mess.setComment(mess.getComment() + Integer.valueOf(valAry[1]));
//                            break;
//                        case LIKES:
//                            mess.setLike(mess.getLike() + Integer.valueOf(valAry[1]));
//                            break;
//                        case VIEWS:
//                            mess.setView(mess.getView() + Integer.valueOf(valAry[1]));
//                            break;
//                    }
//
//                    return JSON.toJSONString(mess);
//                }, Materialized.as("hot-atricle-stream-count-001"))
//                //"COLLECTION:0,COMMENT:0,LIKES:10,VIEWS:20" -> 对象
//                .toStream()
//                .map((key, value) -> new KeyValue<>(key.key(),value))
//                //发送消息
//                .to(HotArticleConstants.HOT_ARTICLE_INCR_HANDLE_TOPIC);
//
//        return stream;
//
//
//    }
//
//}